package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * Author: Felix
 * Date: 2021/9/22
 * Desc: 维度异步关联的函数
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T>{
    private ExecutorService executorService;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //获取线程池对象
        executorService = ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //开启多个线程，发送异步请求，完成维度关联
        executorService.submit(
            new Runnable() {
                @Override
                public void run() {
                    try {
                        long start = System.currentTimeMillis();
                        //run方法中完成的就是维度关联的代码实现
                        //步骤1：根据流中的对象，获取维度的主键
                        String key = getKey(obj);
                        //步骤2：根据主键 到维度表中将维度对象查询出来
                        JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                        //步骤3：将查询出来的维度信息 补充到流中的对象属性上
                        join(obj,dimInfoJsonObj);
                        long end = System.currentTimeMillis();
                        System.out.println("异步维度查询耗时:" + (end - start) +"毫秒");
                        //将异步请求返回结果继续向下游传递
                        resultFuture.complete(Collections.singleton(obj));
                    } catch (Exception e) {
                        e.printStackTrace();
                        throw new RuntimeException("~~~~维度异步关联失败~~~");
                    }
                }
            }
        );
    }
}
